package com.hzy.kafka.service;


import com.alibaba.fastjson.JSON;
import com.hzy.kafka.entity.DelayMessage;
import com.hzy.kafka.entity.MyDelayQueue;
import com.hzy.kafka.enums.KafkaConstants;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.DelayQueue;

@Slf4j
@Service
public class DelayConsumerService {
    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    // 集合
    private static DelayQueue<MyDelayQueue> delayQueue = new DelayQueue<>();

    /**
     * 监听
     * @param json
     * @return
     * @throws Throwable
     */
    @KafkaListener(topics = {KafkaConstants.KAFKA_TOPIC_MESSAGE_DELAY}, containerFactory = "kafkaContainerFactory")
    public boolean onMessage(String json) throws Throwable {
        try {
            DelayMessage delayMessage = JSON.parseObject(json, DelayMessage.class);
            if (!isDelay(delayMessage)) {
                // 如果接收到消息时，消息已经可以发送了，直接发送到实际的队列
                sendActualTopic(delayMessage);
            } else {
                // 存储
                localStorage(delayMessage);
            }
        } catch (Throwable e) {
            log.error("consumer kafka delay message[{}] error!", json, e);
            throw e;
        }
        return true;
    }

    /**
     * 立即执行
     * @param delayMessage
     * @return
     */
    private boolean isDelay(DelayMessage delayMessage) {
        if (delayMessage.getTime().compareTo(0L) == 0){
            return false;
        }
        return true;
    }

    /**
     * 发送消息
     * @param delayMessage
     */
    private void sendActualTopic(DelayMessage delayMessage) {
        kafkaTemplate.send(delayMessage.getActualTopic(), JSON.toJSONString(delayMessage));
    }

    /**
     * 添加集合
     * @param delayMessage
     */
    @SneakyThrows
    private void localStorage(DelayMessage delayMessage) {
        delayQueue.add(new MyDelayQueue(delayMessage));
    }

    /**
     * 加载监听
     */
    @PostConstruct
    private void handleDelayQueue() {
        while (true){
            try {
                if (delayQueue.size() > 0){
                    // 取出队列
                    MyDelayQueue take = delayQueue.take();
                    if (null == take){
                        // 延迟
                        Thread.sleep(1000);

                        continue;
                    }
                    // 将队列发送到队列中
                    DelayMessage delayMessage = take.getDelayMessage();
                    sendActualTopic(delayMessage);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
                log.error("handler kafka rocksdb delay message error!", e);
            }
        }
    }
}
